package com.hliushi.spark.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext, sql}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/17 9:45
 */
class DatasetAndDataFrameIntro {

  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("rdd_intro")

  val sc = new SparkContext(conf)


  /**
   * SparkSQL 中有一个新的入口, 叫做SparkSession
   * SparkSession
   * .      作为整个Spark生态工具的统一入口点, 包括了SQLContext, HiveContext, SparkContext, StreamingContext等组件的功能
   * .      为了支持更多的数据源, 应该完善读取和写入系统
   * .      同时对于原来的入口点也不能放弃, 要向下兼容
   */
  val spark: SparkSession = new sql.SparkSession.Builder()
    .appName("ds_intro")
    .master("local[6]")
    .getOrCreate()

  /**
   *
   */
  @Test
  def rddIntro(): Unit = {
    /**
     * 注意下, Idea为这个函数式编程, 显示了每一步的算子的返回值类型, 这是如何触发呢??? 2021年5月17日10:10:30
     */
    sc.textFile("dataset/wordcount.txt")
      .flatMap(x => x.split(","))
      .map(x => x -> 1)
      .reduceByKey((total, curr) => total + curr)
      .collect()
      .foreach(println(_))

    sc.stop()
  }


  /**
   *
   */
  @Test
  def dsIntro(): Unit = {
    import spark.implicits._

    val sourceRDD = spark.sparkContext.parallelize(Seq(Person("zhangsan", 20), Person("lisi", 15)))

    val personDS = sourceRDD.toDS()

    // 方式1: 通过对象来处理
    personDS.filter(item => item.age > 15).show()
    personDS.filter(_.age > 15).show()

    // 方式2: 通过字段来处理
    val resultDS = personDS.where('age > 10)
      .where('age < 20)
      .select('name)
      .as[String]

    // 方式3: 通过类似SQL的表达式来处理
    personDS.filter("age > 15").show()

    resultDS.show()
  }


  @Test
  def dfIntro(): Unit = {
    import spark.implicits._

    val sourceRDD = spark.sparkContext.parallelize(Seq(Person("zhangsan", 25), Person("lisi", 17)))

    val personDF: DataFrame = sourceRDD.toDF()

    personDF.createOrReplaceTempView("person")

    val resultDF = spark.sql("select name from person where age > 15 and age < 20")

    resultDF.show()
  }

  /**
   *
   *
   */
  @Test
  def dataset(): Unit = {
    // 1.创建SparkSession

    // 2.导入隐式转换
    // 注意: spark在此处不是包, 而是SparkSession对象
    import spark.implicits._
    // 3.演示

    /**
     * Person是一个强类型的类
     * -
     * Dataset是分布式的数据集合
     * Dataset具备强类型的特点,同时支持Lambda函数, 但只能在scala和Java中使用[因为它们是强类型语言???]
     */
    // val value = spark.createDataset(Seq(Person("zhangsan", 25), Person("lisi", 17)))
    val sourceRDD = spark.sparkContext.parallelize(Seq(Person("zhangsan", 25), Person("lisi", 17)))
    // sourceRDD 算子想使用 toDS()/toDF() API 必须导入隐式转换
    val dataset = sourceRDD.toDS()

    // Dataset支持强类型[RDD似]的API
    dataset.filter(item => item.age > 10).show()
    // Dataset支持弱类型API
    dataset.filter('age > 10).show()
    dataset.filter($"age" > 10).show()

    // Dataset可以直接编写SQL表达式
    dataset.filter("age > 10").show()
  }


  /**
   *
   */
  @Test
  def datasetExplain(): Unit = {
    // 1.创建SparkSession
    // 2.导入隐式转换
    import spark.implicits._
    val dataset = spark.createDataset(Seq(Person("zhangsan", 25), Person("lisi", 17)))
    //dataset.explain(true)

    /**
     * 函数原型
     * lazy val toRdd: RDD[InternalRow] = executedPlan.execute()
     * 无论Dataset中放置的是什么类型的对象, 最终执行计划都是RDD[InternalRow]
     */
    val executionRDD: RDD[InternalRow] = dataset.queryExecution.toRdd

    /**
     * == Parsed Logical Plan ==
     * LocalRelation [name#2, age#3]
     *
     * == Analyzed Logical Plan ==
     * name: string, age: int
     * LocalRelation [name#2, age#3]
     *
     * == Optimized Logical Plan ==
     * LocalRelation [name#2, age#3]
     *
     * == Physical Plan ==
     * LocalTableScan [name#2, age#3]
     */
  }

  @Test
  def datasetExplain2(): Unit = {
    // 2.导入隐式转换
    import spark.implicits._
    val dataset = spark.createDataset(Seq(Person("zhangsan", 25), Person("lisi", 17)))

    // 直接获取到已经分析和解析过的Dataset的执行计划, 从中拿到RDD
    val executionRDD: RDD[InternalRow] = dataset.queryExecution.toRdd

    // 通过将Dataset底层的RDD[InternalRow], 通过Decoder转出了和Dataset一样类似的RDD
    val typedRDD = dataset.rdd

    /**
     * (2) MapPartitionsRDD[1] at toRdd at DatasetAndDataFrameIntro.scala:172 []
     * |  ParallelCollectionRDD[0] at toRdd at DatasetAndDataFrameIntro.scala:172 []
     */
    println(executionRDD.toDebugString)

    /**
     * (2) MapPartitionsRDD[5] at rdd at DatasetAndDataFrameIntro.scala:175 []
     * |  MapPartitionsRDD[4] at rdd at DatasetAndDataFrameIntro.scala:175 []
     * |  MapPartitionsRDD[3] at rdd at DatasetAndDataFrameIntro.scala:175 []
     * |  ParallelCollectionRDD[2] at rdd at DatasetAndDataFrameIntro.scala:175 []
     */
    println(typedRDD.toDebugString)

  }


  /**
   * DataFrame是SparkSQL中一个表示关系型数据库中表的函数式抽象
   * --   也就是说DataFrame中有Schema信息, 可以像操作表一样操作 DataFrame
   * --   DataFrame有两部分构成, 一个是row的集合, 每个row对象表示一个行, 二是描述DataFrame结构的Schema
   *
   * 函数原型
   * def where(condition: Column): Dataset[T]
   * def select(cols: Column*): DataFrame
   * def show(numRows: Int): Unit
   */
  @Test
  def dataFrame(): Unit = {
    // 1.创建SparkSession
    // 2.创建DataFrame
    import spark.implicits._
    // 导入了隐式转换, 可以将Seq类型新添加了toDS()/toDF()方法
    val dataFrame = Seq(Person("zhangsan", 15), Person("lisi", 20)).toDF()

    // 3.使用DataFrame
    dataFrame.where('age > 10)
      .select('name)
      .show()

  }

  /**
   * 创建DataFrame的三种方式
   */
  @Test
  def createDataFrame(): Unit = {
    import spark.implicits._
    val personList = Seq(Person("zhangsan", 15), Person("lisi", 20))

    // 1.toDF  --> 使用反射推断
    /**
     * DataFrame是弱类型, Dataset是强类型
     */
    val df1: DataFrame = personList.toDF()
    val ds1: Dataset[Person] = personList.toDS()
    /**
     * 1.1 由RDD创建DataFrame
     * spark支持两种方式把RDD转换为DataFrame, 分别是使用反射推断和指定Schema转换
     * #
     * 调用 toDF() 则转换为 dataFrame
     * 以编程方式指定Schema
     * def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame
     */
    val df2 = spark.sparkContext.parallelize(personList).toDF()

    // 2.createDataFrame
    val df3 = spark.createDataFrame(personList)

    // 3.read
    val df4 = spark.read.csv("dataset/BeijingPM20100101_20151231_noheader.csv")

    /**
     * show() 方法不填写任何参数, 只会显示20条数据
     * 可以用它的重装方法 def show(numRows: Int): Unit
     * def show(): Unit = show(20)
     * only showing top 20 rows
     */
    df4.show()

    /**
     * DataFrame与Dataset互相转换
     */
    val toDS = df4.as[Person]

    val toDF = toDS.toDF()
  }

  /**
   * Row对象表示是 一个行
   */
  @Test
  def row(): Unit = {
    // 1.Row如何创建, 它是什么
    // row对象必须配合Schema对象才会有列名
    val per = Person("zhangsan", 15)
    val row = Row("zhangsan", 15)

    // 2.如何从Row中获取数据
    //println(s"row.getString(0) = ${row.getString(0)}")
    row.getString(1)
    val i1 = row.getInt(2)
    val value = row(1)

    // 获取时可以指定类型
    val i = row.getAs[Int](1)

    // 3.Row也是样例类, 所以也可以使用模式匹配
    row match {
      case obj@Row(name, age) => println(name, age)
    }
  }


  /**
   * dataset 与 dataFrame 的对比
   * #
   * DataFrame就是Dataset[Row], 或者说DataFrame是Dataset[Row]的别名
   * #
   * DataFrame表示的含义是一个支持函数式操作的表, 而Dataset表达是一个类似RDD的东西, Dataset可以处理任何对象
   * DataFrame中所存放的是Row对象, 而Dataset中可以存放任何类型的对象
   * #
   * DataFrame是弱类型, Dataset是强类型
   * #
   * Dataset的操作方式和DataFrame是一样的, 但是对于强类型操作而言, 他们处理的类型不同
   * .    DataFrame在进行强类型操作时候, 例如map算子, 其处理的数据类型永远是Row
   * .    但是对于Dataset来讲, 其中是什么类型, 它就处理什么类型
   * #
   * DataFrame只可以做到运行时类型检测, Dataset做到编译时和运行时都类型检测
   *
   */
  @Test
  def datasetCompDataFrame(): Unit = {
    import spark.implicits._
    val personList = Seq(Person("zhangsan", 15), Person("lisi", 20))

    /**
     * DataFrame是弱类型, Dataset是强类型
     */
    val df: DataFrame = personList.toDF()

    df.map((row: Row) => Row(row.get(0), row.getAs[Int](1) * 2))(RowEncoder.apply(df.schema))
      .show()

    // DataFrame 所代表的弱类型操纵是编译时不完全 [编译时不检测, 运行时才检测]
    // org.apache.spark.sql.AnalysisException: Cannot resolve column name "nameEd" among (name, age);
    df.groupBy("nameEd")

    val ds: Dataset[Person] = personList.toDS()
    ds.map((person: Person) => Person(person.name, person.age * 2))
      .show()

    // Dataset 所代表的操作, 是类型安全的, 编译时安全的 [编译时和运行时都类型检测]
    // ds.filter(person => person.nameEd)
  }


  /**
   * 案例: 根据BeijingPM20100101_20151231.csv 表中year, month进行分组, 统计求和PM_Dongsi, 并按照这个字段排序
   * 使用命令式API, 就是使用DataFrame和Dataset的各种算子去处理
   * #
   * RelationalGroupedDataset类 中 sum方法函数原型
   * def sum(colNames: String*): DataFrame
   */
  @Test
  def dataFrameOp(): Unit = {
    // 0.注意一定要提前导入隐式转换
    //import spark.implicits._
    import spark.implicits._
    // 1.创建SparkSession
    val session = SparkSession.builder()
      .master("local[6]")
      .appName("pm_analysis")
      .getOrCreate()

    // 2.0.指定Schema, 不要自动推断
    // No,year,month,day,hour,season,PM_Dongsi,PM_Dongsihuan,PM_Nongzhanguan,PM_US_Post,DEWP,HUMI,PRES,TEMP,cbwd,Iws,precipitation,Iprec
    val structType = StructType(
      List(
        StructField("id", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("season", IntegerType),
        StructField("PM_Dongsi", DoubleType),
        StructField("PM_Dongsihuan", DoubleType),
        StructField("PM_Nongzhanguan", DoubleType),
        StructField("PM_US_Post", DoubleType),
        StructField("DEWP", IntegerType),
        StructField("HUMI", IntegerType),
        StructField("PRES", IntegerType),
        StructField("TEMP", IntegerType),
        StructField("cbwd", StringType),
        StructField("Iws", DoubleType),
        StructField("precipitation", DoubleType),
        StructField("Iprec", DoubleType)
      )
    )

    // 2.读取数据集  并且去除CSV文件的表头
    val sourceRDD: DataFrame = session.read
      .schema(structType)
      .option("header", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")

    // 查看DataFrame的Schema信息, 要意识到DataFrame中是有结构信息的, 叫做Schema
    sourceRDD.printSchema()

    /**
     * 此处有一个问题 [或许我现在不还知道解决办法, 不一定称之为bug], 使用这种函数式编程方式, 处理数据, 在sum("PM_Dongsi") 这个出算子中
     * -    报错了 :"PM_Dongsi" is not a numeric column. Aggregation function can only be applied on a numeric column.;
     * -    以为这个DataFrame的Schema中的 PM_Dongsi是一个字符串类型, 不能进行 sum运算, [肯定需要特殊转换, 这一点不如下面直接执行SQL来的痛快]
     *
     * 时间:2021年5月19日17:28:10
     * 解决问题, 通过读取数据集的时候,指定Schema, 不要自动推断 option("interSchema", value = true)
     */

    // 3.处理
    //      3.1.选择列
    //      3.2.过滤掉NA的PM记录   使用 na.drop("any", List("PM_Dongsi")) 和 where($"PM_Dongsi" =!= Double.NaN) 效果一样
    //      3.3.分组  select year, month, count(PM_Dongsi) from ... where PM_Dongsi != NA group by year, month
    //      3.4.sum聚合
    //      3.5.把聚合后字段重命名
    //      3.6.根据聚合的字段降序排序
    val result = sourceRDD
      .select('year, 'month, $"PM_Dongsi")
      .na.drop("any", List("PM_Dongsi"))
      .where($"PM_Dongsi" =!= Double.NaN)
      .groupBy($"year", $"month")
      .sum("PM_Dongsi")
      .withColumnRenamed("sum(PM_Dongsi)", "pms")
      .sort($"pms".desc)

    result.printSchema()
    println("-" * 100)
    result.show()

    // 4.关闭资源
    session.stop()
  }

  /**
   * 案例: 根据BeijingPM20100101_20151231.csv 表中year, month进行分组, 统计求和PM_Dongsi, 并按照这个字段排序
   * 使用声明式API, 直接执行SQL去解决这个问题
   */
  @Test
  def dataFrameOp2(): Unit = {
    // 0.注意一定要提前导入隐式转换
    //import spark.implicits._
    // 1.创建SparkSession
    val session = SparkSession.builder()
      .master("local[6]")
      .appName("pm analysis")
      .getOrCreate()

    // 2.读取数据集  并且去除CSV文件的表头
    val sourceRDD: DataFrame = session.read
      .option("inferSchema", value = true)
      .option("header", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")

    // 是否直接使用Sql语句进行查询  --> 肯定是可以的
    // 时间: 2021年5月19日15:22:22  直接写SQL, 在这里并没有考虑类型转换的问题
    //  1.将DataFrame注册为临表
    sourceRDD.createOrReplaceTempView("pm")

    val sql =
      """
        |SELECT
        |   year, month, SUM(PM_Dongsi) AS pms
        |FROM pm
        |WHERE PM_Dongsi != 'NA'
        |GROUP BY year, month
        |ORDER BY pms DESC
        |""".stripMargin

    session.sql(sql).show()
    // 4.关闭资源
    session.stop()
  }
}

/**
 * org.apache.spark.sql.AnalysisException:
 * Unable to generate an encoder for inner class `com.hliushi.spark.sql.Intro$Person` without access to the scope that this class was defined in.
 * Try moving this class out of its parent class.;
 *
 * @param name
 * @param age
 */
case class Person(name: String, age: Int)

case class Student(name: String, age: Int, gpa: Float)

case class PM(year: Int, month: Int, pm_dongsi: Int)

case class StuBean(name: String, alias: String, age: Int, pga: Double, otherInfo: String)